package cn.sheep.dolphin.report

import cn.sheep.dolphin.bean.AreaReportResult
import cn.sheep.dolphin.common.{DolphinAppComm, DolphinKpi}
import cn.sheep.dolphin.config.DolphinConfig
import org.apache.spark.sql.{Row, SQLContext, SaveMode}

/**
  * 地域指标统计 - core
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/11/29
  */
object AreaAnalysisCore {

	def main(args: Array[String]): Unit = {

		if (args.size != 1) {
			println(
				"""
				  |Usage: cn.sheep.dolphin.report.AreaAnalysisCore
				  |	<parquetInputPath>
				""".stripMargin)
			sys.exit(-1)
		}

		val Array(parquetInputPath) = args

		val sc = DolphinAppComm.createSparkContext("地域指标统计")
		val sQLContext = new SQLContext(sc)

		// 读取数据parquet
		val dataFrame = sQLContext.read.parquet(parquetInputPath)

		import sQLContext.implicits._
		dataFrame.map(row => {

			// 获取省市及日期
			val day = row.getAs[String]("requestdate").substring(0, 10)
			val proName = row.getAs[String]("provincename")
			val cityName = row.getAs[String]("cityname")


			((day, proName, cityName),DolphinKpi.caculateKpi(row))
		}).reduceByKey((a, b) => {
			a.zip(b).map(tp => tp._1 + tp._2)
		}).map {
			case ((day, pname, cname), List(rawReq, effReq, adReq, isRTB, succRTB, adShow, adClick, expense, cost)) =>
				AreaReportResult(
					day, pname, cname,
					rawReq.toInt, effReq.toInt, adReq.toInt,
					isRTB.toInt, succRTB.toInt, adShow.toInt,
					adClick.toInt, expense, cost)
		}
			.toDF()
			.write.mode(SaveMode.Append)
			.jdbc(DolphinConfig._url, "r_area_analysis", DolphinConfig.props)

		sc.stop()
	}
}


